
// KafkaProducer.cs
using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace Beauty.Net.Kafka
{
    /// <summary>
    /// Kafka 生产者实现
    /// </summary>
    public class KafkaProducer : IKafkaProducer, IDisposable
    {
        private readonly IProducer<Null, string> _producer;
        private readonly ILogger<KafkaProducer> _logger;
        private readonly KafkaConfig _config;

        public KafkaProducer(KafkaConfig config, ILogger<KafkaProducer> logger)
        {
            _config = config;
            _logger = logger;

            var producerConfig = new ProducerConfig
            {
                BootstrapServers = config.BootstrapServers,
                // 添加事务支持所需的配置
                TransactionalId = config.ProducerConfig.ContainsKey("transactional.id") ? 
                    config.ProducerConfig["transactional.id"] : Guid.NewGuid().ToString(),
                // 确保幂等性
                EnableIdempotence = true,
                // 设置必要的确认级别
                Acks = Acks.All
            };

            // 添加额外配置
            foreach (var kvp in config.ProducerConfig)
            {
                producerConfig.Set(kvp.Key, kvp.Value);
            }

            _producer = new ProducerBuilder<Null, string>(producerConfig).Build();
            
            // 初始化事务
            try
            {
                _producer.InitTransactions(TimeSpan.FromSeconds(10));
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to initialize Kafka transactions");
                throw;
            }
        }

        /// <summary>
        /// 发送单条消息
        /// </summary>
        public async Task<DeliveryResult> ProduceAsync(string topic, IKafkaMessage message)
        {
            try
            {
                var result = await _producer.ProduceAsync(topic, new Message<Null, string>
                {
                    Value = message.Value
                });

                return new DeliveryResult
                {
                    Topic = result.Topic,
                    Partition = (int)result.Partition,
                    Offset = (long)result.Offset,
                    Key = message.Key,
                    IsSuccess = true
                };
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to produce message to topic {Topic}", topic);
                return new DeliveryResult
                {
                    Topic = topic,
                    Key = message.Key,
                    IsSuccess = false,
                    Error = ex.Message
                };
            }
        }

        /// <summary>
        /// 批量发送消息
        /// </summary>
        public async Task<IEnumerable<DeliveryResult>> ProduceBatchAsync(string topic, IEnumerable<IKafkaMessage> messages)
        {
            var tasks = new List<Task<DeliveryResult>>();
            foreach (var message in messages)
            {
                tasks.Add(ProduceAsync(topic, message));
            }

            return await Task.WhenAll(tasks);
        }

        /// <summary>
        /// 事务发送
        /// </summary>
        public async Task<IEnumerable<DeliveryResult>> ProduceTransactionAsync(string topic, IEnumerable<IKafkaMessage> messages)
        {
            try
            {
                _producer.InitTransactions(TimeSpan.FromSeconds(10));
                _producer.BeginTransaction();

                var results = new List<DeliveryResult>();
                foreach (var message in messages)
                {
                    var result = await _producer.ProduceAsync(topic, new Message<Null, string>
                    {
                        Value = message.Value
                    });

                    // 刷新以确保消息被发送到代理
                    _producer.Flush(TimeSpan.FromSeconds(5));

                    results.Add(new DeliveryResult
                    {
                        Topic = result.Topic,
                        Partition = (int)result.Partition,
                        Offset = (long)result.Offset,
                        Key = message.Key,
                        IsSuccess = true
                    });
                }

                _producer.CommitTransaction(TimeSpan.FromSeconds(10));
                return results;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to produce transaction messages to topic {Topic}", topic);
                _producer.AbortTransaction(TimeSpan.FromSeconds(10));
                throw;
            }
        }
        public void Dispose()
        {
            _producer?.Dispose();
        }
    }
}
